package tableapi

import bean.SensorReading
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.table.api.scala.{StreamTableEnvironment, _}
import org.apache.flink.table.api.{DataTypes, EnvironmentSettings, Over, Table, Tumble}
import org.apache.flink.table.descriptors._
import org.apache.flink.types.Row

/**
  * @Description: TODO QQ1667847363
  * @author: xiao kun tai
  * @date:2021/11/27 12:14
  */
object Table10_TimeAndWindow {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    //file source
    val inputPath: String = "src/main/resources/sensor.txt"
    val fileStream: DataStream[String] = env.readTextFile(inputPath)

    val socketStream = env.socketTextStream("192.168.88.106", 7777)

    //先转换为特定的类型
    val dataStream: DataStream[SensorReading] = fileStream.map(data => {
      val arr = data.split(",")
      SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
    })
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(1)) {
        override def extractTimestamp(t: SensorReading): Long = t.timestamp * 1000L
      })

    val settings: EnvironmentSettings = EnvironmentSettings.newInstance()
      .useBlinkPlanner()
      .inStreamingMode()
      .build()

    //创建表执行环境
    val tableEnv = StreamTableEnvironment.create(env, settings)

    val sensorTable: Table = tableEnv.fromDataStream(dataStream, 'id, 'temperature, 'timestamp.rowtime as 'ts)

    //Group Window
    val resultTable: Table = sensorTable
      .window(Tumble over 10.seconds on 'ts as 'tw) //每10秒滚动一次，滚动时间窗口
      .groupBy('id, 'tw)
      .select('id, 'id.count, 'temperature.avg, 'tw.end)
    resultTable.toAppendStream[Row].print("table")

    tableEnv.createTemporaryView("sensor",sensorTable)
    val resultSqlTable:Table = tableEnv.sqlQuery(
      """
        |select
        | id,
        | count(id),
        | avg(temperature),
        | tumble_end(ts,interval '10' second)
        | from sensor
        | group by
        |  id,
        |  tumble(ts,interval '10' second)
      """.stripMargin)

    resultSqlTable.toRetractStream[Row].print("sql")

    //Over Window
    //table api
    val overResultTable: Table = sensorTable
      .window(Over partitionBy 'id orderBy 'ts preceding 2.rows as 'ow)
      .select('id, 'ts, 'id.count over 'ow, 'temperature.avg over 'ow)
    overResultTable.toAppendStream[Row].print("over result")

    //Sql
    val overResultSqlTable:Table = tableEnv.sqlQuery(
      """
        |select
        | id,
        | ts,
        | count(id) over ow,
        | avg(temperature) over ow
        | from sensor
        | window ow as (
        |  partition by id
        |  order by ts
        |  rows between 2 preceding and current row
        | )
      """.stripMargin)

    overResultSqlTable.toRetractStream[Row].print("over sql")

    env.execute("window and time ")
  }


}
